package main

import (
	"sync"
	"net/http"
	"runtime"
	"github.com/sideshow/apns2/certificate"
	"fmt"
	"time"
	"strconv"
	"crypto/tls"
	"golang.org/x/net/http2"
	"github.com/garyburd/redigo/redis"
	"encoding/json"
	"strings"
	"bytes"
)

const maxWorkers = 10000                //最大worker数量
const initWorkers = 2                   //初始化时worker数量
const workerExpire = 100                //worker的最大有效期
const redisQueueKey = "redis_queue_key" //redis的key
const redisQueueNum = 10                //redis队列的数量
const redisHostPort = "127.0.0.1:6379"  //redis服务器
const httpPushSwitch = 1                //1开启http push  其他关闭http push
const httpPushHostPort = "127.0.0.1:8080"
const apnsKeyFilename = "C:/Users/dennis/Desktop/PushChatProAll.pem"
const apnsKeyPassword = ""

var workerPool [5000]workerInfo //每个worker的状态
var m *sync.Mutex                 //创建锁
var chanQueue chan string         //创建通道

type workerInfo struct {
	workerStatus int //woker状态
	workerId     int //worker的id
	expire       int //worker的有效期
}

type PostData struct {
	Alert  string `json:"alert"`
	Badge  int    `json:"badge,string"`
	Sound  string `json:"sound"`
	Tokens string `json:"tokens"`
	Data   string `json:"data"`
}

/**
 * 功能描述：http post方式新增一条推送
 */
func addPush(w http.ResponseWriter, req *http.Request) {
	p := &PostData{}
	req.ParseForm()
	//接受alert参数
	if len(req.Form["alert"]) > 0 {
		p.Alert = req.Form["alert"][0]
	} else {
		w.Write([]byte("{\"code\":-1,\"data\":\"\",\"msg\":\"alert不能为空\"}"))
		return
	}
	if len(req.Form["tokens"]) > 0 {
		p.Tokens = req.Form["tokens"][0]
	} else {
		w.Write([]byte("{\"code\":-1,\"data\":\"\",\"msg\":\"tokens不能为空\"}"))
		return
	}
	//接受badge类型
	if len(req.Form["badge"]) > 0 {
		ret, err := strconv.Atoi(req.Form["badge"][0])
		if err != nil {
			p.Badge = 1
		} else {
			p.Badge = ret
		}
	} else {
		p.Badge = 1
	}
	//处理sound
	if len(req.Form["sound"]) > 0 {
		p.Sound = req.Form["badge"][0]
	} else {
		p.Sound = "default"
	}
	//处理sound
	if len(req.Form["data"]) > 0 {
		p.Data = req.Form["data"][0]
	} else {
		p.Data = "data"
	}

	tokensArray := strings.Split(p.Tokens, "|")
	for i := 0; i < len(tokensArray); i++ {
		if tokensArray[i] != "" {
			p.Tokens = tokensArray[i]
			jsonStr, err := json.Marshal(p)
			if err == nil {
				str := string(jsonStr)
				chanQueue <- string(str)
			} else {
				errorLog("json编码出错")
			}
		}
	}
	size := len(chanQueue)
	w.Write([]byte("{\"code\":0,\"data\":\"" + strconv.Itoa(size) + "\",\"msg\":\"ok\"}"))
}

/**
 * 功能描述：启动http服务
 */
func startHttpServer() {
	if httpPushSwitch == 1 {
		log("start http server " + httpPushHostPort)
		http.HandleFunc("/addPush", addPush)
		http.ListenAndServe(httpPushHostPort, nil)
	}
}

/**
 * 功能描述:推送出错重试
 */
func pushError(urlStr string, data string) {
	c, err := redis.Dial("tcp", redisHostPort)
	if err != nil {
		errorLog("Connect to redis error")
		return
	}
	defer c.Close()
	redisKey := "apns-push-error-num" + urlStr
	errNum, err := redis.Int(c.Do("incr", redisKey))
	if err == nil {
		redis.Int(c.Do("expire", redisKey, 86400))
		if errNum < 10 {
			chanQueue <- string(data)
		}
		errorLog("推送出错:" + strconv.Itoa(errNum) + "," + data + "," + urlStr)
	}
}

/**
 * 功能描述：推送进程  推送消息到苹果服务器
 */
func startWorker() {
	//	var data string
	workerId := getWorkerId()
	if workerId < 0 {
		runtime.Goexit()
	}
	cert, err := certificate.FromPemFile(apnsKeyFilename, apnsKeyPassword)
	if err != nil {
		fmt.Println(err)
		runtime.Goexit()
	}
	client, err := initClient(cert)
	if err != nil {
		fmt.Println(err)
		runtime.Goexit()
	}
	var nowTime int
	nowTime = getNowTimestamp()
	workerPool[workerId].expire = nowTime + workerExpire
	for {
		select {
		case data := <-chanQueue:
			log(data)
			nowTime = getNowTimestamp()
			workerPool[workerId].expire = nowTime + workerExpire
			p := &PostData{}
			err := json.Unmarshal([]byte(data), p)
			if err != nil {
				errorLog("解析json出错" + data)
				continue
			}
			urlStr := fmt.Sprintf("%v/3/device/%v", "https://api.push.apple.com", p.Tokens)
			payload := fmt.Sprintf("{\"aps\":{\"alert\":\"%v\",\"sound\":\"%v\",\"badgeNum\":%v}}", p.Alert, p.Sound, p.Badge)
			//payload := "{\"aps\":{\"alert\":\"" + data + "\",\"sound\":\"msg_high.m4a\",\"badgeNum\":1}}"
			req, err := http.NewRequest("POST", urlStr, bytes.NewReader([]byte(payload)))
			if err != nil {
				pushError(urlStr, data)
				errorLog("apns 推送出错" + payload)
				runtime.Goexit()
			}
			req.Header.Set("Content-Type", "application/json")
			req.Header.Set("apns-topic", "com.foundaoCNNews")
			resp, err := client.Do(req)
			if err != nil {
				pushError(urlStr, data)
				errorLog("workerId=" + strconv.Itoa(workerId) + ",推送出错:" + payload + "," +
					strconv.Itoa(resp.StatusCode))
				runtime.Goexit()
			}
			if resp.StatusCode == http.StatusOK {
				id := resp.Header.Get("apns-id")
				log("workerId=" + strconv.Itoa(workerId) + ",apns-id:" + id + ",http code" + strconv.Itoa(http.StatusOK))
			} else {
				pushError(urlStr, data)
				errorLog("workerId=" + strconv.Itoa(workerId) + "推送错误" + strconv.Itoa(resp.StatusCode) + payload)
				runtime.Goexit()
			}
			resp.Body.Close()
			runtime.Gosched()
		case <-time.After(time.Second * 1):
			nowTime = getNowTimestamp()
			expire := workerPool[workerId].expire - nowTime
			if expire <= 0 {
				log("work_id=" + strconv.Itoa(workerId) + " 退出进程")
				runtime.Goexit()
			}
		}
	}
}

/**
 * 功能描述：初始化进程池参数
 */
func initWorkerPool() {
	for i := 0; i < maxWorkers; i++ {
		workerPool[i].workerStatus = 0
		workerPool[i].expire = 0
		workerPool[i].workerId = 0
	}
}

//获取当前的时间戳
func getNowTimestamp() (int) {
	t := time.Now()
	timestamp := strconv.FormatInt(t.UTC().UnixNano(), 10)
	timestamp = timestamp[:10]
	nowTime, err := strconv.Atoi(timestamp)
	if err != nil {
		return 0
	}
	return nowTime
}

/**
 * 功能描述：获取进程的id
 */
func getWorkerId() (int) {
	m.Lock()
	workerId := 0
	nowTime := getNowTimestamp()
	for i := 0; i < maxWorkers; i++ {
		if workerPool[i].expire < nowTime {
			workerPool[i].expire = nowTime + workerExpire
			workerId = i
			break
		}
	}
	m.Unlock()
	return workerId
}

/**
 * 功能描述：初始化与apns服务器连接
 */
func initClient(cert tls.Certificate) (*http.Client, error) {
	config := &tls.Config{
		Certificates: []tls.Certificate{cert},
	}
	config.BuildNameToCertificate()
	transport := &http.Transport{TLSClientConfig: config}
	if err := http2.ConfigureTransport(transport); err != nil {
		return nil, err
	}
	return &http.Client{Transport: transport}, nil
}

/**
 * 功能描述：日志记录
 */
func log(str string) {
	format := "2006-01-02 15:04:05"
	now := time.Now()
	local1, _ := time.LoadLocation("UTC")
	fmt.Println(now.In(local1).Format(format) + " 【normal】" + str)
}

/**
 * 功能描述：错误日志
 */
func errorLog(str string) {
	format := "2006-01-02 15:04:05"
	now := time.Now()
	local1, _ := time.LoadLocation("UTC")
	fmt.Println(now.In(local1).Format(format) + " 【error】 " + str)
}

func main() {
	log("start app ……")
	go startHttpServer()
	chanQueue = make(chan string, 10000)
	m = new(sync.Mutex)
	initWorkerPool() //初始化进程池
	//初始化redis
	log("init redis connect ……")
	c, err := redis.Dial("tcp", redisHostPort)
	if err != nil {
		errorLog("Connect to redis error")
		return
	}
	defer c.Close()
	//初始化worker
	log("worker total " + strconv.Itoa(initWorkers))
	for i := 0; i < initWorkers; i++ {
		log("init worker index " + strconv.Itoa(i))
		go startWorker()
	}
	redisQueueIndex := 0
	lastChanSize := -1
	total := 0
	for true {
		chanQueueSize := len(chanQueue)
		if chanQueueSize > 0 {
			//队列中的数量大于 50  或者连续  10次队列中的数量没有发送变化  则启动新进程处理
			if chanQueueSize > 20 || total > 20 {
				log("add worker,chan size = " + strconv.Itoa(chanQueueSize) + ",total=" + strconv.Itoa(total))
				//计算需要启动进程的数量
				addWorkerNum := chanQueueSize - 100
				if addWorkerNum < 1 {
					addWorkerNum = 1
				}
				for i := 0; i < addWorkerNum; i++ {
					go startWorker()
				}
				total = 0
				runtime.Gosched()
				time.Sleep(time.Second) //休眠1秒 等待进程处理
			}
			if lastChanSize == chanQueueSize {
				total++
			} else {
				total = 0
			}
			lastChanSize = chanQueueSize
		}
		key := redisQueueKey + strconv.Itoa(redisQueueIndex)
		data, _ := redis.String(c.Do("Rpop", key))
		if data != "" {
			chanQueue <- string(data)
		} else {
			redisQueueIndex++
			if redisQueueIndex >= redisQueueNum {
				redisQueueIndex = 0
				data, err := redis.String(c.Do("ping"))
				if err != nil || data != "PONG" {
					c.Close()
					newRedisClient, err := redis.Dial("tcp", redisHostPort)
					c = newRedisClient
					if err != nil {
						errorLog("reconnect to redis error")
						runtime.Gosched()
						time.Sleep(time.Second)
					}
					log("reconnect to redis success")
				} else {
					runtime.Gosched()
					time.Sleep(time.Microsecond * 300)
				}
			}
			runtime.Gosched()
			time.Sleep(time.Microsecond * 100) //休眠100毫秒
		}
	}
}
